#include "libcomm.h"
#if (COMM_FEATURE_LIBEVENT==1)

class MessageSocketClient : public SocketClient
{
public:
	MessageReactor* m_reactor;
	MessageHandler* m_messageHandler;
	MessageParser* m_messageParser;
	bool m_isOpened;
	MessageSocketClient(MessageReactor* reactor)
		:m_reactor(reactor),m_messageHandler(NULL),m_messageParser(NULL),m_isOpened(false)
	{
		MYLIB_ASSERT_NOTNULL(m_reactor);
		m_messageParser = m_reactor->m_handlerFactory->CreateParser();
		MYLIB_ASSERT_NOTNULL(m_messageParser);
		m_messageHandler = m_reactor->m_handlerFactory->CreateHandler();
		MYLIB_ASSERT_NOTNULL(m_messageHandler);
		m_messageHandler->m_param_opaque = this;
	}

	~MessageSocketClient()
	{
		if(NULL != m_messageHandler)
		{
			m_reactor->m_handlerFactory->DeleteHandler(m_messageHandler);
			m_messageHandler = NULL;
		}
		if(NULL != m_messageParser)
		{
			m_reactor->m_handlerFactory->DeleteParser(m_messageParser);
			m_messageParser = NULL;
		}
	}

	virtual void OnRead(const char* buffer, int size){ MYLIB_ASSERT(m_messageParser->Push(buffer,size));}

	virtual void OnOpen()
	{ 
		m_isOpened = true;
		m_messageHandler->OnOpen();
		m_reactor->m_clientList.Append(this);
	}

	// reason: CONN_EOF(peer shutdown),CONN_ERR(connection error),CONN_SHUTDOWN(close by manual).
	/*can not be blocked.*/
	virtual void OnClose(ConnectionState reason)
	{ 
		if(m_isOpened)
		{
			// remove client.
			m_reactor->m_clientList.Remove(this, false);
			// remove client's message.
			{
				ThreadLockGuard guard_(&m_reactor->m_lockMessage);
				for (MessageReactor::MessageList::iterator itor = m_reactor->m_messageList.begin(); itor != m_reactor->m_messageList.end(); )
				{
					MessageSocketClient* client = (MessageSocketClient*)itor->first;
					if (client == this)
					{
						delete itor->second;
						itor = m_reactor->m_messageList.erase(itor);
						continue;
					}
					itor++;
				}
			}
		}
		m_messageHandler->OnClose(reason);
		m_isOpened = false;
	}
};

MessageHandler::MessageHandler():m_param_opaque(NULL)
{
}

SocketClient* MessageHandler::Peer() const
{
	MYLIB_ASSERT_NOTNULL(m_param_opaque);
	MessageSocketClient* client = (MessageSocketClient*)m_param_opaque;
	return client;
}

MessageParser* MessageHandler::Parser() const
{
	MYLIB_ASSERT_NOTNULL(m_param_opaque);
	MessageSocketClient* client = (MessageSocketClient*)m_param_opaque;
	return client->m_messageParser;
}

bool MessageHandler::Write(const void* buffer, int size)
{
	MYLIB_ASSERT_NOTNULL(m_param_opaque);
	MessageSocketClient* client = (MessageSocketClient*)m_param_opaque;
	Buffer message;
	client->m_messageParser->PackageMessage(message,buffer,size);
	return client->Write(message.get(),message.size());
}

class MessageSocketClientFactory : public SocketClientFactory
{
public:
	MessageReactor* m_reactor;
	MessageSocketClientFactory(MessageReactor* reactor):m_reactor(reactor){}
	virtual SocketClient* Create()
	{
		return new MessageSocketClient(m_reactor);
	}
};

// cyclical parsing every client's input raw buffer.
static int MessageParseProc(CommType& arg)
{
	MessageReactor* reactor = (MessageReactor*)((void*)arg);
	while(reactor->m_handleContinued)
	{
		Thread::Sleep(reactor->m_dispatchIntervalMs);
		{
			ThreadLockGuard guard(reactor->m_clientList.Locker());
			foreach(SocketClient*,client_,reactor->m_clientList.list())
			{
				MessageSocketClient* client = (MessageSocketClient*)client_;
				Buffer message;
				MessageParser::ParseResult result = client->m_messageParser->Parse(message,0);
				switch (result)
				{
				case MessageParser::PR_SUCESS:
					{
						ThreadLockGuard guard_(&reactor->m_lockMessage);
						reactor->m_messageList.push_back(std::make_pair((SocketClient*)client, new Buffer(message)));
					}
					break;
				case MessageParser::PR_ERROR:
					{
						LOGWARN("data package format error on connection \"%s\".",(char*)client->Address().ToString());
						client->Close();
					}
					break;
				default:
					break;
				}
			}
		}

	}
	return 0;
}

// cyclical dispatching every client's input message.
static int MessageDispatchProc(int index, CommType& arg)
{
	(void*)&index;
	MessageReactor* reactor = (MessageReactor*)((void*)arg);
	while(reactor->m_handleContinued)
	{
		MessageSocketClient* client = NULL;
		ScopedPtr<Buffer> messagePtr;
		{
			ThreadLockGuard guard_(&reactor->m_lockMessage);
			MessageReactor::MessageList::iterator itor = reactor->m_messageList.begin();
			if (itor != reactor->m_messageList.end())
			{
				client = (MessageSocketClient*)itor->first;
				client->AddRef();
				messagePtr = itor->second;
				reactor->m_messageList.erase(itor);
			}
		}

		if (messagePtr != NULL)
		{
			MYLIB_ASSERT_NOTNULL(client);
			client->m_messageHandler->OnMessage(messagePtr->get(), messagePtr->size());
			client->Release();
			continue;
		}

		Thread::Sleep(reactor->m_dispatchIntervalMs);
	}
	return 0;
}

MessageReactor::MessageReactor(const SocketAddress& address, MessageHandlerFactory* factory, bool asClient,
							   int dispatchThreadCount, int dispatchIntervalMs,
							   const char* crtfile, const char* keyfile)
	:m_acceptor(NULL),m_connector(NULL),m_parseThread(NULL),m_dispatchThreadPool(NULL),m_dispatchIntervalMs(dispatchIntervalMs),
	m_handlerFactory(factory),m_clientList(new ValueEqualListReleaser<SocketClient*>()),
	m_handleContinued(1)
{
	// add by wanghh 2021.7.30
#ifdef __GNUC__
	//MYLIB_ASSERT(SIG_ERR != signal(SIGPIPE, SIG_IGN));
#endif
	MYLIB_ASSERT_NOTNULL(m_handlerFactory);
	if(asClient)
	{
		(void*)crtfile;
		(void*)keyfile;
		m_connector = new SocketConnector(address,new MessageSocketClient(this));
		MYLIB_ASSERT_NOTNULL(m_connector);
	}
	else
	{
		m_acceptor = new SocketAcceptor(address,new MessageSocketClientFactory(this),crtfile,keyfile);
		MYLIB_ASSERT_NOTNULL(m_acceptor);
	}
	m_parseThread = Thread::Run(MessageParseProc,(void*)this,"msg_parse_trd");
	MYLIB_ASSERT_NOTNULL(m_parseThread);
	m_dispatchThreadPool = ThreadPool::Start(MessageDispatchProc,(void*)this,dispatchThreadCount,"msg_dispatch_pool");
	MYLIB_ASSERT_NOTNULL(m_dispatchThreadPool);
}

MessageReactor::~MessageReactor(void)
{
	Exit();
}

void MessageReactor::Exit()
{
	m_handleContinued = 0;
	// notice all client to exit.
	if(NULL != m_acceptor || NULL != m_connector)
	{
		ThreadLockGuard guard(m_clientList.Locker());
		foreach(SocketClient*,client,m_clientList.list())
			client->Close();
	}

	if(NULL != m_parseThread)
	{
		m_parseThread->Join();
		delete m_parseThread;
		m_parseThread = NULL;
	}
	if(NULL != m_dispatchThreadPool)
	{
		m_dispatchThreadPool->Stop();
		delete m_dispatchThreadPool;
		m_dispatchThreadPool = NULL;
	}

	if(NULL != m_acceptor)
	{
		delete m_acceptor;
		m_acceptor = NULL;
	}
	if(NULL != m_connector)
	{
		delete m_connector;
		m_connector = NULL;
	}

	if(NULL != m_handlerFactory)
	{
		m_handlerFactory->Release();
		m_handlerFactory = NULL;
	}

	{
		ThreadLockGuard guard_(&m_lockMessage);
		for (MessageList::iterator itor = m_messageList.begin(); itor != m_messageList.end(); itor++)
		{
			delete itor->second;
		}
		m_messageList.clear();
	}
}


MessageHandler* MessageReactor::Messager(const char* address)
{
	SocketClient* client_ = NULL;
	if(NULL == address)
	{
		if(NULL != m_connector)
			client_ = m_connector->get();
		if(NULL != m_acceptor)
			m_clientList.Peek(client_);
	}
	else
	{
		ThreadLockGuard guard(m_clientList.Locker());
		foreach(SocketClient*,item,m_clientList.list())
		{
			if(item->Address().ToString().find(address))
			{
				client_ = item;
				break;
			}
		}
	}
	
	if(NULL == client_)
		return NULL;
	MessageSocketClient* client = (MessageSocketClient*)client_;
	return client->m_messageHandler;
}


#endif
